package server

import (
	"fmt"
	"learning/gooop/etcd/raft/common"
	"learning/gooop/etcd/raft/config"
	"learning/gooop/etcd/raft/debug"
	"learning/gooop/etcd/raft/logger"
	"learning/gooop/etcd/raft/lsm"
	rrpc "learning/gooop/etcd/raft/rpc"
	"learning/gooop/etcd/raft/store"
	"net"
	"net/rpc"
	"path"
	"strconv"
	"strings"
	"time"
)

type tRaftKVServer struct {
	mConfig   config.IRaftConfig
	mListener *net.TCPListener
}

func (me *tRaftKVServer) BeginServeTCP(sConfigDir string) error {
	// load store
	logStore := me.loadStore(sConfigDir)

	// load config
	raftCfg, port := me.loadConfig(sConfigDir)
	logger.Logf("tRaftKVServer.BeginServeTCP, starting %v, port=%v", raftCfg.GetID(), port)
	me.mConfig = raftCfg

	// resolve address
	addy, err := net.ResolveTCPAddr("tcp", fmt.Sprintf("0.0.0.0:%d", port))
	if err != nil {
		return err
	}

	// create raft lsm singleton
	raftLSM := lsm.NewRaftLSM(raftCfg, logStore)

	// register raft rpc server
	rpcsrv := rpc.NewServer()
	rs := &RaftRPCServer{
		mRaftLSM: raftLSM,
	}
	err = rpcsrv.Register(rs)
	if err != nil {
		return err
	}

	// register kv rpc server
	ks := &KVStoreRPCServer{
		mRaftLSM: raftLSM,
	}
	err = rpcsrv.Register(ks)
	if err != nil {
		return err
	}

	inbound, err := net.ListenTCP("tcp", addy)
	if err != nil {
		return err
	}

	// start lisening
	me.mListener = inbound
	go rpcsrv.Accept(inbound)
	logger.Logf("tRaftKVServer.BeginServeTCP, service ready at port=%v", port)

	me.whenStartThenWatchDebugKill()
	return nil
}

func (me *tRaftKVServer) whenStartThenWatchDebugKill() {
	go func() {
		for range time.Tick(50 * time.Millisecond) {
			if debug.KilledNodeID == me.mConfig.GetID() {
				logger.Logf("tRaftKVServer.whenStartThenWatchDebugKill, killing %v", me.mConfig.GetID())
				_ = me.mListener.Close()
				return
			}
		}
	}()
}

func (me *tRaftKVServer) Close() error {
	return me.mListener.Close()
}

func (me *tRaftKVServer) loadStore(sConfigDir string) store.ILogStore {
	err, st := store.NewBoltStore(path.Join(sConfigDir, common.DataFileName))
	if err != nil {
		panic(err)
	}
	return st
}

func (me *tRaftKVServer) loadConfig(sConfigDir string) (raftCfg config.IRaftConfig, port int) {
	raftCfg = config.LoadJSONFile(path.Join(sConfigDir, common.ConfigFileName))
	var nodeCfg config.IRaftNodeConfig
	for _, it := range raftCfg.GetNodes() {
		if it.GetID() == raftCfg.GetID() {
			nodeCfg = it
			break
		}
	}
	if nodeCfg == nil {
		panic("node config not found")
	}
	port, err := strconv.Atoi(strings.Split(nodeCfg.GetEndpoint(), ":")[1])
	if err != nil {
		panic(err)
	}

	return raftCfg, port
}

// RaftRPCServer exposes a raft rpc service
type RaftRPCServer struct {
	mRaftLSM lsm.IRaftLSM
}

// Heartbeat leader to follower
func (me *RaftRPCServer) Heartbeat(cmd *rrpc.HeartbeatCmd, ret *rrpc.HeartbeatRet) error {
	e := me.mRaftLSM.Heartbeat(cmd, ret)
	logger.Logf("RaftRPCServer.Heartbeat, cmd=%v, ret=%v, e=%v", cmd, ret, e)
	return e
}

// AppendLog leader to follower
func (me *RaftRPCServer) AppendLog(cmd *rrpc.AppendLogCmd, ret *rrpc.AppendLogRet) error {
	e := me.mRaftLSM.AppendLog(cmd, ret)
	logger.Logf("RaftRPCServer.AppendLog, cmd=%v, ret=%v, e=%v", cmd, ret, e)
	return e
}

// CommitLog leader to follower
func (me *RaftRPCServer) CommitLog(cmd *rrpc.CommitLogCmd, ret *rrpc.CommitLogRet) error {
	e := me.mRaftLSM.CommitLog(cmd, ret)
	logger.Logf("RaftRPCServer.CommitLog, cmd=%v, ret=%v, e=%v", cmd, ret, e)
	return e
}

// RequestVote candidate to others
func (me *RaftRPCServer) RequestVote(cmd *rrpc.RequestVoteCmd, ret *rrpc.RequestVoteRet) error {
	e := me.mRaftLSM.RequestVote(cmd, ret)
	logger.Logf("RaftRPCServer.RequestVote, cmd=%v, ret=%v, e=%v", cmd, ret, e)
	return e
}

// Ping to keep alive
func (me *RaftRPCServer) Ping(cmd *rrpc.PingCmd, ret *rrpc.PingRet) error {
	ret.SenderID = me.mRaftLSM.Config().GetID()
	ret.Timestamp = time.Now().UnixNano()

	logger.Logf("RaftRPCServer.Ping, cmd=%v, ret=%v", cmd, ret)
	return nil
}

// KVStoreRPCServer expose a kv storage service
type KVStoreRPCServer struct {
	mRaftLSM lsm.IRaftLSM
}

// ExecuteKVCmd leader to follower
func (me *KVStoreRPCServer) ExecuteKVCmd(cmd *rrpc.KVCmd, ret *rrpc.KVRet) error {
	e := me.mRaftLSM.ExecuteKVCmd(cmd, ret)
	logger.Logf("KVStoreRPCServer.ExecuteKVCmd, cmd=%v, ret=%v, e=%v", cmd, ret, e)
	return e
}
